package com.example.test.rabbit;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

@Slf4j
@Component
public class RabbitProducer {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private RedisTemplate<String,Object> redisTemplate;

    @PostConstruct
    public void init() {
        rabbitTemplate.setMandatory(true);

        // 入队失败回调
        rabbitTemplate.setReturnsCallback((message) -> {
            log.error("消息推送MQ Queue失败, 请检查路由键准确, 交换机:{}, 路由键:{}", message.getExchange(), message.getRoutingKey());

            String messageId = message.getMessage().getMessageProperties().getMessageId();
            String payload = (String) redisTemplate.opsForHash().get("rabbit.redis.cache", messageId);
            redisTemplate.opsForList().leftPush("rabbit.redis.retry", payload);
            redisTemplate.opsForHash().delete("rabbit.redis.cache", messageId);
        });

        // 交换机签收回调
        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if (ack){
                redisTemplate.opsForHash().delete("rabbit.redis.cache", correlationData.getId());
            }
            else{
                String messageId = correlationData.getId();
                log.error("消息推送MQ Exchange失败, 请检查MQ是否异常, 消息ID: {}", messageId);

                String payload = (String) redisTemplate.opsForHash().get("rabbit.redis.cache", messageId);
                redisTemplate.opsForList().leftPush("rabbit.redis.retry", payload);
                redisTemplate.opsForHash().delete("rabbit.redis.cache", messageId);
            }
        }));
    }

    public void produce(String exchangeName, String routingKey, String payload, String messageId) {
        rabbitTemplate.convertAndSend(exchangeName, routingKey, payload, (message) -> {
            message.getMessageProperties().setMessageId(messageId);
            return message;
        }, new CorrelationData(messageId));
    }
}
